package batch;

import org.apache.rocketmq.common.message.Message;

import java.util.Iterator;
import java.util.List;
import java.util.Map;

public class ListSplitter implements Iterator<List<Message>> {
    private static final int SIZE_LIMIT = 10*1000;
    private final List<Message> messages;
    private  int currentIndex;

    public ListSplitter(List<Message> messages) {
        this.messages = messages;
    }

    @Override
    public boolean hasNext() {
        return currentIndex < messages.size();
    }

    @Override
    public List<Message> next() {
        int nextIndex = currentIndex;
        int totalSize = 0;
        for (;nextIndex < messages.size();nextIndex++){
            Message message = messages.get(nextIndex);
            int messageSize = message.getBody().length + message.getTopic().length();
            Map<String,String> properties = message.getProperties();
            Iterator<Map.Entry<String,String>> iterator = properties.entrySet().iterator();
            while(iterator.hasNext()){
                messageSize+=iterator.next().getKey().length()+iterator.next().getValue().length();
            }
            messageSize = messageSize +20;
            if (messageSize>SIZE_LIMIT){
                //第一次就超出了限制，直接跳过这条消息继续扫描
                if (nextIndex - currentIndex ==0){
                    nextIndex++;
                }
                break;
            }
            //如果当前发送列表已经超出限制，暂停添加消息
            if (messageSize + totalSize>SIZE_LIMIT){
                break;
            }else{
                totalSize+=messageSize;
            }
        }
        List<Message> messages = this.messages.subList(currentIndex,nextIndex);
        currentIndex=nextIndex;
        return messages;
    }

}
